package com.demo1.Listener;

import com.demo1.util.RedisStreamUtil;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.connection.stream.RecordId;
import org.springframework.data.redis.connection.stream.StreamInfo;
import org.springframework.data.redis.stream.StreamListener;
import org.springframework.stereotype.Component;

import java.util.Map;
import java.util.Objects;

/**
 *
 */
@Slf4j
@Component
@RequiredArgsConstructor
public class ReportReadMqListener implements StreamListener<String, MapRecord<String, String, String>> {


    private final RedisStreamUtil redisStreamUtil;
    private static final String REDPACKAGE="\"redPackage\"";
    private static final String NOTICE="\"notice\"";
    private static final String LOG="\"log\"";
    /***
     *“财务服务处理收到的红包”是一个核心操作，而其他两个操作是辅助性的，
     * 那么可以先确保财务服务操作成功完成后，再发送通知和日志记录消息到队列。这样可以确保核心业务逻辑的优先级和数据一致性。
     * 同时，可以通过异步处理通知和日志记录来提高系统的整体响应性和吞吐量。
     */

    @Override
    public void onMessage(MapRecord<String, String, String> message) {
        // stream的key值
        String streamKey = message.getStream();
        //消息ID
        RecordId recordId = message.getId();
        //消息内容
        Map<String, String> msg = message.getValue();
        //TODO 处理逻辑
        String type=msg.get("type");
        System.out.println(type);

        if(Objects.equals(type, REDPACKAGE)){
            System.out.println("微信钱包增加多少钱"+msg.get("redPackage"));
        } else if (Objects.equals(type, NOTICE)) {
            System.out.println("消息通知用户"+msg.get("notice"));
        } else if (Objects.equals(type, LOG)) {
            System.out.println("log"+msg.get("log"));
        }


        //财务服务处理收到的红包
        //微信钱包消息通知
        //用户消息日志记录
        log.info("【streamKey】= " + streamKey + ",【recordId】= " + recordId + ",【msg】=" + msg);
        //逻辑处理完成后，ack消息，删除消息，group为消费组名称
        StreamInfo.XInfoGroups xInfoGroups = redisStreamUtil.queryGroups(streamKey);
        xInfoGroups.forEach(xInfoGroup -> redisStreamUtil.ack(streamKey, xInfoGroup.groupName(), recordId.getValue()));
        redisStreamUtil.del(streamKey, recordId.getValue());
    }
    /***
     *消息队列通常用于异步处理场景，其核心价值在于解耦生产者和消费者、提高系统吞吐量、增强系统的伸缩性和健壮性。在很多情况下，消息一旦被消费者处理，就不需要将处理结果返回给生产者，因为生产者可能并不关心消息的处理结果，或者处理结果可以通过其他方式获取。

     * 以下是消息队列适用于异步处理但不需要返回信息的一些典型场景：

     * 日志收集：生产者生成日志消息并发送到队列，消费者异步地将日志消息写入到日志存储系统中，生产者不需要知道日志是否成功写入。

     * 订单处理：电商平台中的订单服务将订单消息发送到队列，后续的支付服务、库存服务、发货服务分别作为消费者异步处理这些订单，订单生成服务不需要等待这些后续处理完成。

     * 数据转换和批处理：生产者发送原始数据到队列，消费者异步地进行数据清洗、转换和聚合等操作，然后将处理后的数据存储到数据库或数据仓库中，生产者不需要即时获取处理结果。

     * 通知和邮件发送：用户触发了某个事件（如注册、购买）后，生产者将通知消息发送到队列，消费者异步地发送邮件或短信给用户，通知消息的发送状态不需要反馈给用户。

     * 缓存更新：生产者将更新操作发送到队列，消费者异步地更新缓存数据，以保证缓存的一致性，生产者不需要知道缓存是否已经更新。

     * 在这些场景中，消息队列的使用模式通常是“火和忘记”（fire-and-forget），生产者发送消息后立即继续处理其他任务，而不需要等待消费者的响应。这种模式简化了生产者的逻辑，因为它们不需要处理复杂的错误处理或重试机制
     */
}
